package com.jrx.anytxn.param.job.bactchlimitadjust.config;

import com.jrx.anytxn.param.config.AnytxnBatchProperties;
import com.jrx.anytxn.param.constant.Constant;
import com.jrx.anytxn.param.entity.BtLimitAdjustInfo;
import com.jrx.anytxn.param.entity.TlBatchAdjustLimit;
import com.jrx.anytxn.param.job.bactchlimitadjust.bean.LimitAdjustBean;
import com.jrx.anytxn.param.job.bactchlimitadjust.step.BatchLimitAdjustStep1Processor;
import com.jrx.anytxn.param.job.bactchlimitadjust.step.BatchLimitAdjustStep1Writer;
import com.jrx.anytxn.param.job.bactchlimitadjust.step.BatchLimitAdjustStep2Processor;
import com.jrx.anytxn.param.job.bactchlimitadjust.step.BatchLimitAdjustStep2Writer;
import org.apache.ibatis.session.SqlSessionFactory;
import org.mybatis.spring.batch.MyBatisCursorItemReader;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.support.SynchronizedItemStreamReader;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;

import javax.annotation.Resource;
import java.util.HashMap;
import java.util.Map;

/**
 * 批量调额job
 *
 * @author : zhenghaopu
 * 2020/1/6
 */
@Configuration
@EnableConfigurationProperties(AnytxnBatchProperties.class)
public class BatchLimitAdjustConfiguration {

    @Resource(name = "paramSqlSessionFactory")
    SqlSessionFactory sqlSessionFactory;

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    @Autowired
    private AnytxnBatchProperties anytxnBatchProperties;

    @Autowired
    private BatchLimitAdjustStep1Processor batchLimitAdjustStep1Processor;

    @Autowired
    private BatchLimitAdjustStep1Writer batchLimitAdjustStep1Writer;

    @Autowired
    private BatchLimitAdjustStep2Processor batchLimitAdjustStep2Processor;

    @Autowired
    private BatchLimitAdjustStep2Writer batchLimitAdjustStep2Writer;

    @Autowired
    private PlatformTransactionManager platformTransactionManager;


    @Bean
    public Job batchLimitAdjustJob(){
        return jobBuilderFactory.get("batchLimitAdjustJob").start(batchLimitAdjustStep1()).next(batchLimitAdjustStep2()).build();

    }

    @Bean
    public Step batchLimitAdjustStep1(){
        return stepBuilderFactory.get("batchLimitAdjustStep1")
                .<TlBatchAdjustLimit, LimitAdjustBean>chunk(anytxnBatchProperties.getBatchLimitAdjustConfiguration().getChunkLimit())
                //表示一批处理100
                // 条记录，数据是逐条发送给processor处理的，处理100条记录后，统一返回给writer进行保存处理
                .reader(batchLimitAdjustStep1Reader())
                .processor(batchLimitAdjustStep1Processor)
                .writer(batchLimitAdjustStep1Writer)
                .transactionManager(platformTransactionManager)
                .taskExecutor(new SimpleAsyncTaskExecutor())
                .throttleLimit(anytxnBatchProperties.getBatchLimitAdjustConfiguration().getThrottleLimit())//多线程*/
                .build();
    }

    @Bean
    @StepScope
    public SynchronizedItemStreamReader<TlBatchAdjustLimit> batchLimitAdjustStep1Reader() {

        MyBatisCursorItemReader<TlBatchAdjustLimit> cursorItemReader = new MyBatisCursorItemReader<>();
        cursorItemReader.setSaveState(false);
        cursorItemReader.setSqlSessionFactory(sqlSessionFactory);
        cursorItemReader.setQueryId("com.jrx.anytxn.param.mapper.ext.ExtTlBatchAdjustLimitMapper.selectByHandleStatusForBatch");
        Map<String, Object> map = new HashMap<>();
        map.put("handleStatus", Constant.HANDLE_STATUS_0);
        cursorItemReader.setParameterValues(map);

        SynchronizedItemStreamReader<TlBatchAdjustLimit> synchronizedItemStreamReader = new SynchronizedItemStreamReader<>();
        synchronizedItemStreamReader.setDelegate(cursorItemReader);
        return synchronizedItemStreamReader;
    }

    @Bean
    public Step batchLimitAdjustStep2(){
        return stepBuilderFactory.get("batchLimitAdjustStep2")
                .<BtLimitAdjustInfo, BtLimitAdjustInfo>chunk(anytxnBatchProperties.getBatchLimitAdjustConfiguration().getChunkLimit())
                //表示一批处理100
                // 条记录，数据是逐条发送给processor处理的，处理100条记录后，统一返回给writer进行保存处理
                .reader(batchLimitAdjustStep2Reader())
                .processor(batchLimitAdjustStep2Processor)
                .writer(batchLimitAdjustStep2Writer)
                .transactionManager(platformTransactionManager)
                .taskExecutor(new SimpleAsyncTaskExecutor())
                .throttleLimit(anytxnBatchProperties.getBatchLimitAdjustConfiguration().getThrottleLimit())//多线程*/
                .build();
    }

    @Bean
    @StepScope
    public SynchronizedItemStreamReader<BtLimitAdjustInfo> batchLimitAdjustStep2Reader() {

        MyBatisCursorItemReader<BtLimitAdjustInfo> cursorItemReader = new MyBatisCursorItemReader<>();
        cursorItemReader.setSaveState(false);
        cursorItemReader.setSqlSessionFactory(sqlSessionFactory);
        cursorItemReader.setQueryId("com.jrx.anytxn.param.mapper.ext.ExtBtLimitAdjustInfoMapper.selectListByMqStatusForBatch");
        Map<String, Object> map = new HashMap<>();
        map.put("mqStatus", Constant.MQ_SATTUS_0);
        cursorItemReader.setParameterValues(map);

        SynchronizedItemStreamReader<BtLimitAdjustInfo> synchronizedItemStreamReader = new SynchronizedItemStreamReader<>();
        synchronizedItemStreamReader.setDelegate(cursorItemReader);
        return synchronizedItemStreamReader;
    }
}
